package com.eric.log.mapred;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.eric.log.mapred.LogBasicMapRed.BasicMapper;

public class VisitMapRed {
	public static int sessionCounter = 0;
	
	public static class VisitsReducer extends Reducer<Text,Text,Text,Text> {
		public void reduce (Text key,Iterable<Text> value,Context context) throws IOException, InterruptedException {
			sessionCounter++;
			
			Date MaxTime = null;
			Date entranceTime = null;
			Date logTimeStamp = null;

			String url = null;
			String ref = null;
//			String ua = null;
			String ip = null;
			String[] goal_urls = {Config.goal1,Config.goal2,Config.goal3,Config.goal4,Config.goal5};
			
			int pv = 0;
			long timeOfVisit = 0;

			boolean bounce = false;
			boolean goal = false;
			
			try {
				MaxTime = new SimpleDateFormat(Config.DateFormat,Locale.ENGLISH).parse(Config.MAXDate);
			} catch (ParseException e1) {
				e1.printStackTrace();
			}
			
			entranceTime = MaxTime;
			
			for (Text v : value) {
				pv++;
				
				
				String[] entry = v.toString().replaceAll("\"", "").split(Config.FS);
				try {
					logTimeStamp = new SimpleDateFormat(Config.DateFormat,Locale.ENGLISH).parse(entry[Config.FN_DATE]+" "+entry[Config.FN_TIME]);
					
					//Parse visit entrance
					if (logTimeStamp.compareTo(entranceTime) <= 0 ) {
						entranceTime = logTimeStamp;
						ip = entry[Config.FN_IP];
						url = entry[Config.FN_URL];
						ref = entry[Config.FN_REF];
//						ua = entry[Config.FN_UA];
					}
					
					//Check if visit hit goal
					for (int i=0;i<goal_urls.length;i++) {
						if (!goal_urls[i].equals("") && entry[Config.FN_URL].contains(goal_urls[i])){
							goal = true;
						}
					}
					
					//Parse time of visit
					if (pv > 0) {
						timeOfVisit = logTimeStamp.getTime() - entranceTime.getTime();
					}
					
				} catch (ParseException e) {
					e.printStackTrace();
				}
			}
			if (pv == 1) bounce = true;
			Text newValue = new Text(
				ip+Config.FS+
				entranceTime.toString()+Config.FS+
				url+Config.FS+
				ref+Config.FS+
//				ua+Config.LOG_FS+
				pv+Config.FS+
				bounce+Config.FS+
				goal+Config.FS+
				timeOfVisit);
			
			context.write(key, newValue);
		}
	}
	
	public static void run (String input,String output) throws IOException, ClassNotFoundException, InterruptedException {
		LogBasicMapRed.setKeyFieldNum(Config.FN_UTM);
		
		Configuration conf = new Configuration();
	    Job job = new Job(conf,"Visit Map Reduce");
	    
	    job.setMapperClass(BasicMapper.class);
	    job.setReducerClass(VisitsReducer.class);
	    
	    job.setJarByClass(VisitMapRed.class);
	    
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    
	    FileInputFormat.addInputPath(job, new Path(input));
	    FileOutputFormat.setOutputPath(job, new Path(output));
	    job.waitForCompletion(true);
	    
	    System.out.println("Total visit count:"+sessionCounter);
	}
	  
	public static void main(String[] args) throws Exception {
	    
	}
}